package com.atguigu.chapter11;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @ClassName: Flink14_Function_Scalar
 * @Description:
 * @Author: kele
 * @Date: 2021/4/14 0:09
 *
 * 自定义Table函数(表值函数，一进多出)
 *  1、
 *      -- 输入: 一个或多个
 *      -- 输出: 多个
 *
 *  2、自定义函数extends
 *      添加注解：将一列数据炸成多列，通过该注解给每列添加字段名
 *          -- @FunctionHint(output = @DataTypeHint("Row<word string,len int>"))
 *          -- 该函数没有返回值，通过collect收集输出
 *
 **/
//将一行数据按照，分割，转化为多行
public class Flink15_Function__Table {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        DataStreamSource<String> stream = env.fromElements("hello,world", "java,python", "scala,C,C++");

        Table table = tEnv.fromDataStream(stream, $("words"));

        /**
         * table
         *      1、内联方式
         *      2、注册方式
         */

        //内联方式
/*        table
                //炸出来一张新表，新表会自动和旧表进行连接
                .joinLateral(call(MyFunction.class,$("words")))
                .leftOuterJoinLateral(call(MyFunction.class,$("words")))
                .select($("word"),$("len") )
                .execute()
                .print();*/

        //注册方式
      /*  tEnv.createTemporaryFunction("Fsplit",MyFunction.class);

        table
                .joinLateral(call("Fsplit",$("words")))
                .select($("words"),$("word"),$("len"))
                .execute()
                .print();*/


        /**
         * SQL方式
         */

        tEnv.createTemporaryFunction("Fsplit",MyFunction.class);
        tEnv.createTemporaryView("mtable",table);


        /**
         * select xxx from xxx join lateral table(自定义函数(参数)) on 连接条件
         */
        //内连接
       /* tEnv.sqlQuery("select words,word,len " +
                "from mtable " +
                "join lateral table(Fsplit(words)) on true").execute().print();*/

        //左连接
        /*tEnv.sqlQuery("select words,word,len " +
                "from mtable " +
                "left join lateral table(Fsplit(words)) on true").execute().print();*/

        //将字段起别名
        tEnv.sqlQuery("select words,n_word,n_len " +
                "from mtable " +
                "join lateral table(Fsplit(words)) as T(n_word,n_len)" +
                "on true").execute().print();
    }

    /**
     * 自定义函数：按照,分割，将每行数据转化成（单词，长度）
     *
     * 在自定义的函数中必须实现一个eval的方法
     * 泛型：输出行的类型，
     *      如果炸成多列，可以使用元组，row的类型
     *      使用collect进行输出，返回值是void
     *
     *官网解释：
     *https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/functions/udfs.html#table-functions
     */

    @FunctionHint(output = @DataTypeHint("Row<word string,len int>"))
    public static class MyFunction extends TableFunction<Row> {

        //自定义的eval函数，在使用时会自动调用，使用时会直接将参数传入，并将该函数的返回值返回
        public void eval(String line){

            for (String word : line.split(",")) {

               collect(Row.of(word,word.length()));
            }

        }

    }

}
